
#include <iostream>
#include <fstream>
#include <map>
#include <vector>
#include <string>

#include <boost/asio.hpp>

#include <xbase_utility.hpp>
#include <dynamic_buffer.hpp>

#include "imphub_dbp_protocol.hpp"


	const int _hex_rmap[] = {
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1, // 4 x 8
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1,
		 0,  1,  2,  3,  4,  5,  6,  7, // '0'-'9'
		 8,  9, -1, -1, -1, -1, -1, -1, // 8 x 8
		-1, 10, 11, 12, 13, 14, 15, -1, // 'A'-'F'
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1, // 12 x 8
		-1, 10, 11, 12, 13, 14, 15, -1, // 'a'-'f'
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1,
		-1, -1, -1, -1, -1, -1, -1, -1 // 16 x 8
	};

void hex2bin(std::string &bin, const std::string &hex)
{
	for(std::size_t i=0; i+1<hex.size(); i+=2)
	{
		bin += static_cast<char>((_hex_rmap[hex[i] & 0x7F] << 4) | (_hex_rmap[hex[i+1] & 0x7F]));
	}
}


struct insert_item
{
	std::string m_cid;
	std::string m_peerid;
	std::string m_gcid;
	uint64_t m_filesize;
	uint64_t m_timestamp;

	std::istream &load(std::istream &in) {
		in >> m_cid >> m_peerid
		   >> m_gcid >> m_filesize
		   >> m_timestamp;
		return in;
	}
};

typedef std::map<uint64_t,	// timestamp
		 insert_item> insert_item_map;

typedef std::map<std::string,	// cid
		 basic_packet<insert_request> > request_map;

typedef std::vector< basic_packet<insert_request> > request_vector;

// void sort_data(request_map &request, const insert_item_map &data)
// {
// 	for(insert_item_map::const_iterator it = data.begin();
// 	    it != data.end();
// 	    ++it)
// 	{
// 		insert_request &req = request[it->m_cid];
// 		if(req.m_cid.empty()) // it first inserted
// 		{
// 			req.m_cid = it->m_cid;
// 			req.m_gcid = it->m_gcid;
// 			req.m_filesize = it->m_filesize;
// 		}
// 		req.m_peerid_list.push_back(it->m_peerid);
// 	}
// }

// class row_mutator_pool
// 	: private boost::noncopyable
// {
// public:
// 	row_mutator_pool(xbase::client_environment &env,
// 			 std::size_t max_size = 10000)
// 		: m_env(env)
// 		, m_max_size(max_size) {}

// 	xbase::row_mutator_ptr new_row_mutator() {
// 		return (m_free_mutator.empty()
// 			? m_env.new_row_mutator()
// 			: m_free_mutator.pop_back());
// 	}

// 	void delete_row_mutator(xbase::row_mutator_ptr ptr) {
// 		if(m_free_mutator.size() < m_max_size)
// 			m_free_mutator.push_front(ptr);
// 	}

// private:
// 	xbase::client_environment &m_env;
// 	std::deque<xbase::row_mutator_ptr> m_free_mutator;
// 	std::size_t m_max_size;
// };

// class row_mutation_pool
// 	: private boost::noncopyable
// {
// public:
// 	row_mutator_pool(xbase::protocol::table_info_ptr table,
// 			 std::size_t max_size = 10000)
// 		: m_table_info(table)
// 		, m_max_size(max_size) {}

// 	xbase::row_mutation new_row_mutation(const std::string &row_name) {
// 		if(m_free_mutation.empty())
// 		{
// 			return xbase::row_mutation(m_table_info, row_name);
// 		}
// 		else
// 		{
// 			xbase::row_mutation_data_ptr ptr = m_free_mutation.pop_back();
// 			ptr->change_row_name(row_name);
// 			return xbase::row_mutation(m_table_info, ptr);
// 		}
// 	}

// 	void delete_row_mutation(xbase::row_mutation mutation) {
// 		if(m_free_mutation.size() < m_max_size)
// 		{
// 			mutation.clear();
// 			m_free_mutation.push_front(mutation.data_ptr());
// 		}
// 	}

// private:
// 	xbase::protocol::table_info_ptr m_table_info;
// 	std::deque<xbase::row_mutation_data_ptr> m_free_mutation;
// 	std::size_t m_max_size;
// };

// std::size_t count_commit = 0;
// std::size_t count_complete = 0;
// std::size_t count_failed = 0;

// void on_complete(row_mutator_pool *mutator_pool,
// 		 xbase::row_mutator_ptr mutator,
// 		 row_mutation_pool *mutation_pool,
// 		 xbase::row_mutation mutation,
// 		 const boost::system::error_code &ec)
// {
// 	++count_complete;
// 	if(ec) ++count_failed;
// 	mutator_pool.delete_row_mutator(mutator);
// 	mutation_pool.delete_row_mutation(mutation);
// }

// void insert_data(xbase::client_environment &env,
// 		 row_mutator_pool &mutator_pool,
// 		 row_mutation_pool &mutation_pool,
// 		 const request_map &request)
// {
// 	const std::string column("gcid:_");

// 	for(request_map::const_iterator it = request.begin();
// 	    it != request.end(); ++ it)
// 	{
// 		const insert_request &req = it->second;

// 		std::string cid_data;
// 		hex2bin(cid_data, req.m_cid);

// 		std::string value_data;
// 		hex2bin(value_data, req.m_gcid);
// 		value_data.append(reinterpret_cast<const char*>(&req.m_filesize),
// 				  sizeof(req.m_filesize));

// 		for(std::vector<std::string>::iterator it_p = req.m_peerid_list.begin();
// 		    it_p != req.m_peerid_list.end(); ++ it_p)
// 		{
// 			const std::string row_data = cid_data + *it_p;
// 			xbase::row_mutation mutation = mutation_pool.new_row_mutation(row_data);
// 			mutation.add(column, value_data);

// 			xbase::row_mutator_ptr mutator = mutator_pool.new_row_mutator();
// 			++count_commit;
// 			env.async_apply(mutator, mutation,
// 					boost::bind(on_complete,
// 						    &mutator_pool, mutator,
// 						    &mutation_pool, mutation,
// 						    _1));
// 		}
// 	}
// }

template<typename container>
struct getter;

template<>
struct getter<request_map>
{
	static inline
	const basic_packet<insert_request> &get(const request_map::const_iterator &it) {
		return it->second;
	}
};

template<>
struct getter<request_vector>
{
	static inline
	const basic_packet<insert_request> &get(const request_vector::const_iterator &it) {
		return *it;
	}
};
	

template<typename container>
int insert_data(xbase::simple_buffer &buffer,
		boost::asio::ip::tcp::socket &sock,
		const container &request)
{
	for(typename container::const_iterator it = request.begin();
	    it != request.end(); ++ it)
	{
		const basic_packet<insert_request> &req = getter<container>::get(it);
				
		packet::oarchive out(buffer.prepare(1024 * 1024 * 200));
		if(! req.save(out))
		{
			std::cerr << "encode failed: " << req.to_string() << std::endl;
			return -1;
		}
		buffer.commit(out.used_buffer_size());
		boost::system::error_code ec;
		boost::asio::write(sock, buffer.data(), boost::asio::transfer_all(), ec);
		buffer.clear();
		if(ec)
		{
			std::cerr << "send request failed: " << ec.message() << std::endl;
			return -1;
		}
	}
	return 0;
}

int main(int argc, char **argv)
{
	if((argc < 4) || (argc > 8))
	{
		std::cout << "imphub_dbp_test_insert <dbp ip> <dbp port> <data file>"
			  << " [verbos/nonverbos] [batch_mode/single_mode]"
			  << " [max_item_count=1,000,000] [max_request_count=100,000]" << std::endl;
		return 1;
	}

	const boost::asio::ip::tcp::endpoint ep(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);
	std::ifstream infile(argv[3]);
	// const char *table_name = argv[4];

	boost::asio::io_service ios;
	boost::asio::ip::tcp::socket sock(ios);
	boost::system::error_code ec;
	sock.connect(ep, ec);
	if(ec)
	{
		std::cout << "connect to " << ep << " failed: "
			  << ec.message() << std::endl;
		return 1;
	}

	request_map request;
	request_vector request_vec;
	std::size_t counter = 0;
	std::size_t total_counter = 0;
	std::size_t total_request_counter = 0;
	uint64_t t_total = 0;

	// xbase::client_options_ptr options(new xbase::client_options);
	// xbase::io_service_pool_ptr pool(new xbase::impl::single_io_service);
	// xbase::client_environment env(options, pool, master_ep);

	// xbase::table_info_cache::sync_handler_type sh;
	// env.async_open_table(table_name, sh.make_handler());
	// if(sh.get_error_code())
	// {
	// 	std::cerr << "open table failed: " << sh.get_error_code().message() << std::endl;
	// 	return 0;
	// }
	// xbase::protocol::table_info_ptr table_info = sh.get_table_info();

	// row_mutator_pool mutator_pool(env);
	// row_mutation_pool mutation_pool(table_info);

	bool verbos = true;
	bool single_mode = false;
	std::size_t max_item_count = 1000000;
	std::size_t max_request_count = 100000;
	if(argc >= 5)
	{
		verbos = (std::string("verbos") == argv[4]);
	}
	if(argc >= 6)
	{
		single_mode = (std::string("single_mode") == argv[5]);
	}
	if(argc >= 7)
	{
		max_item_count = ::atoi(argv[6]);
	}
	if(argc >= 8)
	{
		max_request_count = ::atoi(argv[7]);
	}
	if(single_mode)
	{
		request_vec.reserve(max_request_count + 10000);
	}

	xbase::simple_buffer buffer;
	buffer.prepare(1024 * 1024 * 256);

	int ret = 0;

	insert_item item;
	basic_packet<insert_request> p;
	item.m_filesize = bad_filesize;
	while((! infile.eof()) || (! request.empty()))
	{
		if(! infile.eof())
		{
			item.load(infile);

			if(item.m_filesize != bad_filesize)
			{
				if(single_mode)
				{
					p.body().m_cid.clear();
					p.body().m_gcid.clear();
					hex2bin(p.body().m_cid, item.m_cid);
					hex2bin(p.body().m_gcid, item.m_gcid);
					p.body().m_filesize = item.m_filesize;
					p.body().m_peerid_list.clear();
					p.body().m_peerid_list.push_back(item.m_peerid);
					request_vec.push_back(p);
				}
				else
				{
					insert_request &req = request[item.m_cid].body();
					if(req.m_cid.empty()) // it first inserted
					{
						hex2bin(req.m_cid, item.m_cid);
						p.body().m_gcid.clear();
						hex2bin(req.m_gcid, item.m_gcid);
						req.m_filesize = item.m_filesize;
					}
					req.m_peerid_list.push_back(item.m_peerid);
				}
				++ counter;
				++ total_counter;
				item.m_filesize = bad_filesize;
			}
		}

		if((counter >= max_item_count) ||
		   (request.size() >= max_request_count) ||
		   (request_vec.size() >= max_request_count) ||
		   (infile.eof() && ((! request.empty()) || (! request_vec.empty()))))
		{
			if(verbos)
				std::cout << "start inserting ..." << std::endl;
			
			const uint64_t t_start = cur_micro_seconds();

			if(single_mode)
			{
				ret = insert_data(buffer, sock, request_vec);
			}
			else
			{
				ret = insert_data(buffer, sock, request);
			}
			
			const uint64_t t_end = cur_micro_seconds();
			const uint64_t t_used = t_end - t_start;
			const double used = t_used / (1000.0 * 1000.0);
			const std::size_t req_count = (single_mode ? request_vec.size() : request.size());

			t_total += t_used;

			if(ret < 0)
			{
				std::cerr << "insert data failed" << std::endl;
				break;
			}
			else if(verbos)
			{
				std::cout << req_count << " requests with "
					  << counter << " items inserted, used "
					  << used << " seconds"
					  << std::endl;
			}

			total_request_counter += (single_mode ? request_vec.size() : request.size());
			counter = 0;
			request.clear();
			request_vec.clear();
		}
	}

	std::cout << "--- --- ---" << std::endl << "totally "
		  << total_request_counter << " requests with "
		  << total_counter << " items inserted, used "
		  << (t_total / (1000.0 * 1000.0)) << " seconds"
		  << std::endl;

	// 必须等一会再关闭socket，这是dbp的BUG
	sleep(3);
	sock.close();

	return ret;
}

